//! ## Command Queue Between Clients and Server in CS2 System
//!
//! Global command queue is for clients to send commands to server. Clients can push
//! commands to the queue, and server can pop commands from the queue.
//!
//! Backend implementation of command queue is based on POSIX message queue.
//!
//!```txt
//! Clients                                             Server
//! |                                                   ^
//! v               POSIX Message Queue                 |
//! +------------+------------+------------+------------+
//! | Command 1  | Command 2  | Command 3  | Command 4  |
//! +------------+------------+------------+------------+
//! ```
//!
//! ### Commands
//!
//! The command is a struct with a magic number, a client ID, and a command kind.
//! Available command kinds are:
//! - [`CommandKind::Connect`]: Connect to the server.
//! - [`CommandKind::Disconnect`]: Disconnect from the server.
//! - [`CommandKind::PathsAttach`]: Attach paths to the server. To attach the cache
//!   table, clients should first send series of [`CommandKind::PathsAttach`] to inform
//!   the server of the paths to attach. The server will create cache tables for the
//!   paths.
//! - [`CommandKind::Synchronize`]: Synchronize the cache to the server. To synchronize
//!   the cache to the server, the client sends a [`CommandKind::Synchronize`] command to
//!   the server.
//! - [`CommandKind::Terminate`]: Terminate the server.
//!
//! Only connected clients can send [`CommandKind::PathsAttach`] &
//! [`CommandKind::Synchronize`] commands.

use std::{collections::BTreeSet, error, ffi, fmt::Display, mem, process, ptr, result, slice, vec};

use crate::{
    cache::{CacheTableKey, CacheTableKeyRef, CacheTableVal},
    error::{Error, Result},
    utils::{map::Key, retlibc},
};

/// Magic number for the command.
pub(crate) const COMMAND_MAGIC: u32 = 0xC0ADAC7F;

/// Prefix for the POSIX message queue of command queue.
pub const COMMAND_QUEUE_PREFIX: &str = "cs2s-cmdq-";

/// Command queue handle for clients.
///
/// The client should first [`CommandQueueClientHandle::connect`] to the server, then
/// send other commands.
///
/// Drop the handle will close the message queue, and **automatically** disconnect from
/// the server.
pub struct CommandQueueClientHandle {
    queue: CommandQueue,
    connected: bool,
    #[cfg(test)]
    client_id: u32,
}

impl CommandQueueClientHandle {
    /// Create a new command queue for client. The client can send commands to the
    /// server.
    ///
    /// Drop the command queue will close the message queue and disconnect from the
    /// server.
    pub fn new(name: &str) -> Result<Self> {
        CommandQueue::new(name, CommandQueueSide::Client).map(|queue| Self {
            queue,
            connected: false,
            #[cfg(test)]
            client_id: 0,
        })
    }

    /// Connect to the server.
    pub fn connect(&mut self) -> Result<()> {
        let command = Command::new(CommandKind::Connect);

        #[cfg(test)]
        let command = command.with_client_id(self.client_id);

        self.queue.send(command).map(|_| self.connected = true)
    }

    /// Attach paths to the server.
    pub fn paths_attach(&self, paths: impl IntoIterator<Item = impl AsRef<str>>) -> Result<()> {
        self.check_connected()?;

        let paths = paths
            .into_iter()
            .map(|path| path.as_ref().bytes().collect::<Vec<_>>())
            .collect::<Vec<_>>()
            .join(&b':');

        let command = Command::new(CommandKind::PathsAttach {
            paths_len: paths.len(),
        });

        #[cfg(test)]
        let command = command.with_client_id(self.client_id);

        self.queue
            .send(command.with_payload(self.queue.mq_msgsize, [paths])?)
    }

    /// Synchronize the cache to the server.
    pub fn synchronize(
        &self,
        path: impl AsRef<str>,
        cache_key: &CacheTableKeyRef,
        cache_val: &CacheTableVal,
    ) -> Result<()> {
        self.check_connected()?;

        let command = Command::new(CommandKind::Synchronize {
            path_len: path.as_ref().len(),
            cache_key: CacheTableKey::from_ref(cache_key),
            host_meta_size: cache_val.host_meta.len(),
            host_code_size: cache_val.host_code.len(),
        });

        #[cfg(test)]
        let command = command.with_client_id(self.client_id);

        self.queue.send(command.with_payload(
            self.queue.mq_msgsize,
            [
                path.as_ref().as_bytes(),
                cache_val.host_meta,
                cache_val.host_code,
            ],
        )?)
    }

    /// Terminate the server.
    pub fn terminate(name: &str, status: TerminateStatus) -> Result<()> {
        let client = CommandQueue::new(name, CommandQueueSide::Client).map(|queue| Self {
            queue,
            connected: false,
            #[cfg(test)]
            client_id: 0,
        })?;

        let command = Command::new(CommandKind::Terminate { status });

        #[cfg(test)]
        let command = command.with_client_id(client.client_id);

        client.queue.send(command)
    }
}

impl CommandQueueClientHandle {
    /// Disconnect from the server.
    pub(crate) fn disconnect(&self) -> Result<()> {
        let command = Command::new(CommandKind::Disconnect);

        #[cfg(test)]
        let command = command.with_client_id(self.client_id);

        self.queue.send(command)
    }
}

impl CommandQueueClientHandle {
    fn check_connected(&self) -> Result<()> {
        if !self.connected {
            #[cfg(test)]
            let pid = self.client_id;

            #[cfg(not(test))]
            let pid = process::id();

            return Err(Error::InvalidClient { pid });
        }
        Ok(())
    }
}

impl Drop for CommandQueueClientHandle {
    fn drop(&mut self) {
        // Disconnect from the server.
        if self.connected {
            let _ = self.disconnect();
        }

        // Safety: `self.queue.mqd` is a valid message queue descriptor.
        unsafe { libc::mq_close(self.queue.mqd) };
    }
}

/// Command queue handle for server.
///
/// Drop the handle will close and **automatically** unlink the message queue.
pub struct CommandQueueServerHandle {
    queue: CommandQueue,
    clients: BTreeSet<u32>,
    buffer: Box<[u8]>,
}

impl CommandQueueServerHandle {
    /// Create a new command queue for server. The server can receive commands from
    /// multiple clients.
    ///
    /// Drop the command queue will close and unlink the message queue.
    pub fn new(name: &str) -> Result<Self> {
        let queue = CommandQueue::new(name, CommandQueueSide::Server)?;
        let buffer = vec![0u8; queue.mq_msgsize].into_boxed_slice();

        Ok(Self {
            queue,
            clients: BTreeSet::new(),
            buffer,
        })
    }

    /// Get the connected clients.
    pub fn connected_clients(&self) -> impl Iterator<Item = &u32> {
        self.clients.iter()
    }

    /// Listen to the command queue and handle the commands.
    ///
    /// The listener should implement the [`CommandQueueListener`] trait. Some types
    /// (e.g., closures, tuples of closures) already implement the trait.
    ///
    /// Clients should first send a `Connect` command to the server, then send other
    /// commands, and finally send a `Disconnect` command to the server. The server will
    /// check the client ID for each command. Only the connected clients can send other
    /// commands.
    ///
    /// When receiving a [`CommandKind::Terminate`] command, the server will stop.
    pub fn listen(
        &mut self,
        mut listener: impl CommandQueueListener,
    ) -> result::Result<TerminateStatus, Box<dyn error::Error>> {
        loop {
            // Receive a command from the client.
            let msg = self.queue.receive(&mut self.buffer)?;

            // Parse the command.
            let (command, payload) = Command::from_raw(msg)?;

            // Check the command kind.
            match command.kind() {
                CommandKind::Terminate { status } => return Ok(*status),
                CommandKind::Connect => {
                    // Connect the client.
                    self.clients.insert(command.client());
                }
                CommandKind::Disconnect => {
                    // Disconnect the client.
                    self.clients.remove(&command.client());
                }
                _ => {
                    // Check if the client is connected.
                    if !self.clients.contains(&command.client()) {
                        return Err(Into::into(Error::InvalidClient {
                            pid: command.client(),
                        }));
                    }
                }
            }

            // Handle the command.
            listener.handle(command, payload)?;

            // Report the status of the command queue.
            let status = CommandQueueStatus::new(&self.queue)?;
            listener.report(status);
        }
    }
}

impl Drop for CommandQueueServerHandle {
    fn drop(&mut self) {
        // Safety: `self.queue.mqd` is a valid message queue descriptor.
        unsafe {
            libc::mq_close(self.queue.mqd);
            libc::mq_unlink(self.queue.mq_name.as_ptr());
        }
    }
}

/// Status of a command queue.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CommandQueueStatus {
    capacity: usize,
    current: usize,
}

impl CommandQueueStatus {
    /// Get the capacity of the command queue.
    pub fn capacity(&self) -> usize {
        self.capacity
    }

    /// Get the current load of the command queue.
    pub fn current(&self) -> usize {
        self.current
    }
}

impl CommandQueueStatus {
    /// Construct a [`CommandQueueStatus`] instance from a given [`CommandQueue`].
    fn new(queue: &CommandQueue) -> Result<Self> {
        // Safety: The operations calling libc functions in this `unsafe` block are safe.
        // - `mq_getattr` is safe because the message queue descriptor is valid and the
        //   `mq_attr` is a valid pointer to the message queue attributes.
        unsafe {
            let mut attr = mem::zeroed();
            retlibc::check!(libc::mq_getattr(queue.mqd, &mut attr));
            Ok(Self {
                capacity: attr.mq_maxmsg as usize,
                current: attr.mq_curmsgs as usize,
            })
        }
    }
}

/// Listener for the command queue.
pub trait CommandQueueListener {
    /// Handle the command.
    fn handle(&mut self, command: &Command, payload: Vec<&[u8]>) -> Result<()>;

    /// Report the status of the command queue.
    fn report(&mut self, status: CommandQueueStatus);
}

/// Implement the [`CommandQueueListener`] trait for a handler without status report.
impl<H> CommandQueueListener for H
where
    H: FnMut(&Command, Vec<&[u8]>) -> Result<()>,
{
    fn handle(&mut self, command: &Command, payload: Vec<&[u8]>) -> Result<()> {
        self(command, payload)
    }

    fn report(&mut self, _: CommandQueueStatus) {
        // Do nothing.
    }
}

/// Implement the [`CommandQueueListener`] trait for a handler with status report.
impl<H, R> CommandQueueListener for (H, R)
where
    H: FnMut(&Command, Vec<&[u8]>) -> Result<()>,
    R: FnMut(CommandQueueStatus),
{
    fn handle(&mut self, command: &Command, payload: Vec<&[u8]>) -> Result<()> {
        self.0(command, payload)
    }

    fn report(&mut self, status: CommandQueueStatus) {
        self.1(status)
    }
}

/// Command queue.
struct CommandQueue {
    mq_name: ffi::CString,
    mqd: libc::mqd_t,
    mq_msgsize: usize,
}

impl CommandQueue {
    /// Open the message queue.
    fn new(name: &str, side: CommandQueueSide) -> Result<Self> {
        // Evaluate the name.
        let mq_name = Self::evaluate_mq_name(name)?;

        // Safety: The operations calling libc functions in this `unsafe` block are safe.
        // - `mq_open` is safe because the name is a valid C string and the mode is
        //   correct.
        // - `mq_getattr` is safe because the message queue descriptor is valid and the
        //   `mq_attr` is a valid pointer to the message queue attributes.
        let (mqd, mq_msgsize) = unsafe {
            let mqd = match side {
                CommandQueueSide::Client => {
                    retlibc::check!(libc::mq_open(mq_name.as_ptr(), libc::O_WRONLY))
                }
                CommandQueueSide::Server => {
                    retlibc::check!(libc::mq_open(
                        mq_name.as_ptr(),
                        libc::O_RDONLY | libc::O_CREAT | libc::O_EXCL,
                        0o600,
                        ptr::null::<libc::mq_attr>(),
                    ))
                }
            };
            let mut mq_attr = mem::zeroed();
            retlibc::check!(libc::mq_getattr(mqd, &mut mq_attr));
            (mqd, mq_attr.mq_msgsize as usize)
        };

        Ok(Self {
            mq_name,
            mqd,
            mq_msgsize,
        })
    }

    /// Evaluate the name and return a C string.
    /// - The name must be a valid C string.
    fn evaluate_mq_name(name: &str) -> Result<ffi::CString> {
        let name = format!("/{}{}", COMMAND_QUEUE_PREFIX, name);
        ffi::CString::new(name.clone()).map_err(|_| Error::InvalidName { name })
    }
}

impl CommandQueue {
    /// Send a command to the server.
    fn send(&self, command: impl AsRef<[u8]>) -> Result<()> {
        // Safety: The operations calling libc functions in this `unsafe` block are safe.
        // - `mq_send` is safe because the message queue descriptor is valid and the
        //   `command` is a valid pointer to the command.
        unsafe {
            retlibc::check!(libc::mq_send(
                self.mqd,
                command.as_ref().as_ptr() as *const _ as _,
                command.as_ref().len(),
                0,
            ));
        }
        Ok(())
    }

    /// Receive a command from the client.
    ///
    /// The buffer should be large enough (i.e., at least `mq_msgsize`) to hold the
    /// command.
    fn receive<'a>(&self, buffer: &'a mut [u8]) -> Result<&'a [u8]> {
        // Safety: The operations calling libc functions in this `unsafe` block are safe.
        // - `mq_receive` is safe because the message queue descriptor is valid, the
        //   `buffer` is a valid pointer to the message buffer and large enough, the
        //   `msg_size` is the maximum size of the message.
        let len = unsafe {
            retlibc::check!(libc::mq_receive(
                self.mqd,
                buffer.as_mut().as_mut_ptr() as _,
                self.mq_msgsize,
                ptr::null_mut(),
            ))
        } as usize;

        Ok(&buffer[..len])
    }
}

/// Command queue role.
///
/// This is used to determine the access mode of the message queue and shared memory
/// object.
#[repr(C)]
enum CommandQueueSide {
    Client,
    Server,
}

/// Command sent from client to server.
///
/// For commands with payloads, use `with_payload` method to encode the command.
#[repr(C)]
#[derive(Debug, Clone)]
pub struct Command {
    magic: u32,
    client: u32,
    kind: CommandKind,
}

impl Command {
    /// Create a new command.
    pub fn new(kind: CommandKind) -> Self {
        Self {
            magic: COMMAND_MAGIC,
            client: process::id(),
            kind,
        }
    }

    /// Get the client ID.
    #[inline]
    pub fn client(&self) -> u32 {
        self.client
    }

    /// Get the command kind.
    #[inline]
    pub fn kind(&self) -> &CommandKind {
        &self.kind
    }

    /// Parse the raw command.
    ///
    /// Returns error if the `raw` is too short or the magic number is invalid.
    pub fn from_raw(raw: &[u8]) -> Result<(&Self, Vec<&[u8]>)> {
        if raw.len() < mem::size_of::<Command>() {
            return Err(Error::InvalidCommand {
                msg: format!("command is too short: {}", raw.len()),
            });
        }
        // Safety: `raw` is a valid slice of `Command`.
        let command = unsafe { &*(raw.as_ptr() as *const Command) };
        if command.magic != COMMAND_MAGIC {
            return Err(Error::InvalidMagic {
                actual: command.magic,
                expected: COMMAND_MAGIC,
            });
        }
        let payload = match &command.kind {
            CommandKind::Connect | CommandKind::Disconnect | CommandKind::Terminate { .. } => {
                vec![]
            }
            CommandKind::PathsAttach { paths_len } => {
                let joined_paths = Self::payload(raw, [*paths_len])?[0];
                joined_paths.split(|&b| b == b':').collect()
            }
            CommandKind::Synchronize {
                path_len,
                host_meta_size,
                host_code_size,
                ..
            } => Self::payload(raw, [*path_len, *host_meta_size, *host_code_size])?,
        };
        Ok((command, payload))
    }

    /// Encode the command with payloads.
    pub fn with_payload<L, P>(&self, msg_size_max: usize, payloads: L) -> Result<Box<[u8]>>
    where
        L: IntoIterator<Item = P>,
        P: AsRef<[u8]>,
    {
        let payloads = payloads.into_iter().collect::<Vec<_>>();
        let required = payloads.iter().map(|p| p.as_ref().len()).sum::<usize>();
        let free = msg_size_max - mem::size_of::<Command>();
        if required > free {
            return Err(Error::MemNotEnough { required, free });
        }

        let mut data = Vec::with_capacity(mem::size_of::<Command>() + required);
        data.extend_from_slice(self.as_ref());
        for payload in payloads {
            data.extend_from_slice(payload.as_ref());
        }
        Ok(data.into_boxed_slice())
    }
}

impl Command {
    fn payload(raw: &[u8], payload_size: impl AsRef<[usize]>) -> Result<Vec<&[u8]>> {
        let total = payload_size.as_ref().iter().sum::<usize>();

        if raw.len() - mem::size_of::<Command>() != total {
            return Err(Error::InvalidCommand {
                msg: format!(
                    "command payload size mismatch: {}, payloads ({}={}) expected",
                    raw.len() - mem::size_of::<Command>(),
                    payload_size
                        .as_ref()
                        .iter()
                        .map(|s| s.to_string())
                        .collect::<Vec<_>>()
                        .join("+"),
                    total,
                ),
            });
        }

        let mut payloads = &raw[mem::size_of::<Command>()..];
        let mut result = Vec::with_capacity(payload_size.as_ref().len());
        for &size in payload_size.as_ref() {
            let (payload, rest) = payloads.split_at(size);
            result.push(payload);
            payloads = rest;
        }
        Ok(result)
    }

    #[cfg(test)]
    #[inline]
    fn with_client_id(mut self, client: u32) -> Self {
        self.client = client;
        self
    }

    /// Get the base address of the command.
    #[inline]
    fn base(&self) -> *const u8 {
        self as *const _ as _
    }
}

impl Display for Command {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Command from #{}: {}", self.client, self.kind)
    }
}

impl AsRef<[u8]> for Command {
    fn as_ref(&self) -> &[u8] {
        // Safety: `self` is a valid `Command` instance.
        unsafe { slice::from_raw_parts(self.base(), mem::size_of::<Command>()) }
    }
}

/// Command kind.
#[repr(C)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CommandKind {
    Connect,
    Disconnect,
    PathsAttach {
        paths_len: usize,
    },
    Synchronize {
        path_len: usize,
        cache_key: CacheTableKey,
        host_meta_size: usize,
        host_code_size: usize,
    },
    Terminate {
        status: TerminateStatus,
    },
}

impl Display for CommandKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Connect => write!(f, "Connect"),
            Self::Disconnect => write!(f, "Disconnect"),
            Self::PathsAttach { paths_len } => {
                write!(f, "PathsAttach ( paths-len: {} )", paths_len,)
            }
            Self::Synchronize {
                path_len,
                cache_key,
                host_meta_size,
                host_code_size,
            } => {
                let CacheTableKeyRef {
                    guest_addr,
                    guest_size,
                    guest_sign,
                } = cache_key.as_ref();

                write!(
                    f,
                    "Synchronize ( path: {} B; guest code / sign: @{:#X}-{:#X} / {}; host meta / code: {} B / {} B)",
                    path_len, guest_addr, guest_addr + guest_size, guest_sign, host_meta_size, host_code_size,
                )
            }
            Self::Terminate { status } => write!(f, "Terminate ( {} )", status),
        }
    }
}

#[repr(C)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TerminateStatus {
    Success,
    Failure,
}

impl Display for TerminateStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Success => write!(f, "Success"),
            Self::Failure => write!(f, "Failure"),
        }
    }
}

#[cfg(test)]
mod test {
    use std::thread;

    use coverage_helper::test;
    use cs2s_testdata::*;

    use crate::sign::CodeSign;

    use super::*;

    /// Convenient OK-only listener for testing.
    impl CommandQueueListener for Result<()> {
        fn handle(&mut self, _: &Command, _: Vec<&[u8]>) -> Result<()> {
            self.clone()
        }

        fn report(&mut self, _: CommandQueueStatus) {
            // Do nothing.
        }
    }

    #[test]
    fn basic_collaboration() {
        const NAME: &str = "test-basic-collaboration";

        // Create a server.
        let server = CommandQueueServerHandle::new(NAME);
        assert!(server.is_ok());
        let mut server = server.unwrap();

        // Create a client.
        let client = CommandQueueClientHandle::new(NAME);
        assert!(client.is_ok());
        let mut client = client.unwrap();

        let client_handle = thread::spawn(move || {
            // Send commands from client to server.
            assert!(client.connect().is_ok());

            assert!(client.paths_attach(PATHS).is_ok());

            for (i, ((path, guest_code), (host_meta, host_code))) in PATHS
                .iter()
                .zip(GUEST_CODES.iter())
                .zip(HOST_METAS.iter().zip(HOST_CODES.iter()))
                .enumerate()
            {
                assert!(client
                    .synchronize(
                        path,
                        &CacheTableKeyRef {
                            guest_addr: 0x400000 + i * 0x40,
                            guest_size: guest_code.len(),
                            guest_sign: &guest_code.into(),
                        },
                        &CacheTableVal {
                            host_meta,
                            host_code,
                        },
                    )
                    .is_ok());
            }

            assert!(CommandQueueClientHandle::terminate(NAME, TerminateStatus::Success).is_ok());
        });
        let server_handle = thread::spawn(move || {
            // Listen to the commands from client.
            let mut index = 0;
            let index_max = 3 + GUEST_CODES.len();
            let result = server.listen(|command: &Command, payload: Vec<&[u8]>| {
                assert!(index < index_max);
                assert_eq!(command.client(), 0);
                match index {
                    0 => {
                        assert_eq!(command.kind(), &CommandKind::Connect);
                        assert!(payload.is_empty());
                    }
                    1 => {
                        assert_eq!(
                            command.kind(),
                            &CommandKind::PathsAttach {
                                paths_len: PATHS.iter().map(|s| s.len()).sum::<usize>()
                                    + PATHS.iter().count()
                                    - 1,
                            }
                        );
                        assert_eq!(
                            payload[..],
                            PATHS.iter().map(|s| s.as_bytes()).collect::<Vec<_>>()
                        );
                    }
                    i if i > 1 && i < index_max - 1 => {
                        let j = i - 2;
                        assert_eq!(
                            command.kind(),
                            &CommandKind::Synchronize {
                                path_len: PATHS[j].len(),
                                cache_key: CacheTableKey::from_ref(&CacheTableKeyRef {
                                    guest_addr: 0x400000 + j * 0x40,
                                    guest_size: GUEST_CODES[j].len(),
                                    guest_sign: &GUEST_CODES[j].into(),
                                }),
                                host_meta_size: HOST_METAS[j].len(),
                                host_code_size: HOST_CODES[j].len(),
                            }
                        );
                        assert_eq!(
                            payload[..],
                            [PATHS[j].as_bytes(), HOST_METAS[j], HOST_CODES[j]],
                        );
                    }
                    _ => {
                        assert_eq!(command.kind(), &CommandKind::Disconnect);
                        assert!(payload.is_empty());
                    }
                }
                index += 1;
                Ok(())
            });
            assert!(result.is_ok());
        });

        // Wait for the threads to finish.
        assert!(client_handle.join().is_ok());
        assert!(server_handle.join().is_ok());
    }

    #[test]
    fn multiclients() {
        const NAME: &str = "test-multiclients";
        // Create a server.
        let server = CommandQueueServerHandle::new(NAME);
        assert!(server.is_ok());
        let mut server = server.unwrap();

        // Create clients.
        let clients = (0..GUEST_CODES.len()).map(|_| CommandQueueClientHandle::new(NAME));
        assert_eq!(clients.len(), GUEST_CODES.len());
        let clients = clients
            .into_iter()
            .enumerate()
            .map(|(i, result)| {
                assert!(result.is_ok());
                let mut client = result.unwrap();
                client.client_id = i as _;
                client
            })
            .collect::<Vec<_>>();

        // Send commands from clients to server.
        let clients_handle = thread::spawn(move || {
            for (i, (mut client, (path, (guest_code, (host_meta, host_code))))) in clients
                .into_iter()
                .zip(
                    PATHS.iter().zip(
                        GUEST_CODES
                            .iter()
                            .zip(HOST_METAS.iter().zip(HOST_CODES.iter())),
                    ),
                )
                .enumerate()
            {
                assert!(client.connect().is_ok());
                assert!(client
                    .synchronize(
                        path,
                        &CacheTableKeyRef {
                            guest_addr: 0x400000 + i * 0x40,
                            guest_size: guest_code.len(),
                            guest_sign: &guest_code.into(),
                        },
                        &CacheTableVal {
                            host_meta,
                            host_code,
                        }
                    )
                    .is_ok());
            }

            assert!(CommandQueueClientHandle::terminate(NAME, TerminateStatus::Success).is_ok());
        });

        let server_handle = thread::spawn(move || {
            // Listen to the commands from client.
            let mut index = 0;
            let index_max = 3 * GUEST_CODES.len();
            let result = server.listen((
                |command: &Command, payload: Vec<&[u8]>| {
                    assert!(index < index_max);
                    match index {
                        i if i % 3 == 0 => {
                            assert_eq!(command.kind(), &CommandKind::Connect);
                            assert!(payload.is_empty());
                        }
                        i if i % 3 == 1 => {
                            let j = i / 3;
                            assert_eq!(
                                command.kind(),
                                &CommandKind::Synchronize {
                                    path_len: PATHS[j].len(),
                                    cache_key: CacheTableKey::from_ref(&CacheTableKeyRef {
                                        guest_addr: 0x400000 + j * 0x40,
                                        guest_size: GUEST_CODES[j].len(),
                                        guest_sign: &GUEST_CODES[j].into(),
                                    }),
                                    host_meta_size: HOST_METAS[j].len(),
                                    host_code_size: HOST_CODES[j].len(),
                                }
                            );
                            assert_eq!(
                                payload[..],
                                [PATHS[j].as_bytes(), HOST_METAS[j], HOST_CODES[j]],
                            );
                        }
                        _ => {
                            assert_eq!(command.kind(), &CommandKind::Disconnect);
                            assert!(payload.is_empty());
                        }
                    }
                    index += 1;
                    Ok(())
                },
                |status: CommandQueueStatus| {
                    assert!(status.capacity() >= status.current());
                },
            ));
            assert!(result.is_ok());
            assert!(server.connected_clients().count() == 0);
        });

        // Wait for the threads to finish.
        assert!(clients_handle.join().is_ok());
        assert!(server_handle.join().is_ok());
    }

    #[test]
    fn invalid_name_error() {
        // Invalid name with null character.
        assert!(matches!(
            CommandQueueServerHandle::new("test-invalid\0-name"),
            Err(Error::InvalidName { .. })
        ));
        assert!(matches!(
            CommandQueueClientHandle::new("test-invalid\0-name"),
            Err(Error::InvalidName { .. })
        ));

        // Invalid name with slash character.
        assert!(matches!(
            CommandQueueServerHandle::new("test/invalid-name"),
            Err(Error::LibcError {
                errno: libc::EACCES
            })
        ));
        assert!(matches!(
            CommandQueueClientHandle::new("test/invalid-name"),
            Err(Error::LibcError {
                errno: libc::EACCES
            })
        ));
    }

    #[test]
    fn payload_too_large() {
        const NAME: &str = "test-payload-too-large";

        // Create a server.
        let server = CommandQueueServerHandle::new(NAME);
        assert!(server.is_ok());

        // Create a client.
        let client = CommandQueueClientHandle::new(NAME);
        assert!(client.is_ok());
        let mut client = client.unwrap();

        // Send a command from client to server.
        assert!(client.connect().is_ok());

        let long_path = "/lib".repeat(0x2000) + "/libtest.so";
        assert!(matches!(
            client.paths_attach([&long_path]),
            Err(Error::MemNotEnough { .. }),
        ));

        assert!(matches!(
            client.synchronize(
                long_path,
                &CacheTableKeyRef {
                    guest_addr: 0x400000,
                    guest_size: GUEST_CODE_0.len(),
                    guest_sign: &GUEST_CODE_0.into(),
                },
                &CacheTableVal {
                    host_meta: &HOST_META_0,
                    host_code: &HOST_CODE_0,
                },
            ),
            Err(Error::MemNotEnough { .. }),
        ));

        assert!(matches!(
            client.synchronize(
                PATH_0,
                &CacheTableKeyRef {
                    guest_addr: 0x400000,
                    guest_size: GUEST_CODE_0.len(),
                    guest_sign: &GUEST_CODE_0.into(),
                },
                &CacheTableVal {
                    host_meta: &HOST_META_0,
                    host_code: &[0x5F; 0x2000],
                },
            ),
            Err(Error::MemNotEnough { .. }),
        ));
    }

    #[test]
    fn message_size_error() {
        const NAME: &str = "test-message-size-error";

        // Create a server.
        let server = CommandQueueServerHandle::new(NAME);
        assert!(server.is_ok());

        // Create a client.
        let client = CommandQueueClientHandle::new(NAME);
        assert!(client.is_ok());
        let client = client.unwrap();

        assert!(matches!(
            Command::from_raw(vec![0u8; mem::size_of::<Command>() - 1].as_slice()),
            Err(Error::InvalidCommand { msg }) if msg == format!(
                "command is too short: {}",
                mem::size_of::<Command>() - 1,
            ),
        ));
        let actual = Command::from_raw(
            &Command::new(CommandKind::Synchronize {
                path_len: PATH_0.len(),
                cache_key: CacheTableKey::from_ref(&CacheTableKeyRef {
                    guest_addr: 0x400000,
                    guest_size: GUEST_CODE_0.len(),
                    guest_sign: &GUEST_CODE_0.into(),
                }),
                host_meta_size: HOST_META_0.len(),
                host_code_size: HOST_CODE_0.len() + 1,
            })
            .with_payload(
                client.queue.mq_msgsize,
                [PATH_0.as_bytes(), &HOST_META_0, &HOST_CODE_0],
            )
            .unwrap(),
        )
        .unwrap_err();
        let expect = Error::InvalidCommand {
            msg: format!(
                "command payload size mismatch: {}, payloads ({}+{}+{}={}) expected",
                PATH_0.len() + HOST_META_0.len() + HOST_CODE_0.len(),
                PATH_0.len(),
                HOST_META_0.len(),
                HOST_CODE_0.len() + 1,
                PATH_0.len() + HOST_META_0.len() + HOST_CODE_0.len() + 1,
            ),
        };
        assert_eq!(actual, expect);
    }

    #[test]
    fn message_corruption() {
        const NAME: &str = "test-message-corruption";

        // Create a server.
        let server = CommandQueueServerHandle::new(NAME);
        assert!(server.is_ok());
        let mut server = server.unwrap();

        // Create a client.
        let client = CommandQueueClientHandle::new(NAME);
        assert!(client.is_ok());
        let mut client = client.unwrap();

        let client_handle = thread::spawn(move || {
            // Send a good command from client to server.
            assert!(client.connect().is_ok());

            // Send a bad command from client to server.
            assert!(client.queue.send([0xFF; mem::size_of::<Command>()]).is_ok());
        });

        let server_handle = thread::spawn(move || {
            // Listen to the commands from client.
            let result = server
                .listen(Ok(()))
                .map_err(|error| error.downcast_ref::<Error>().cloned().unwrap());
            assert!(matches!(result, Err(Error::InvalidMagic { .. }),));
        });

        // Wait for the threads to finish.
        assert!(client_handle.join().is_ok());
        assert!(server_handle.join().is_ok());
    }

    #[test]
    fn client_not_connected_error() {
        const NAME: &str = "test-client-not-connected";

        // Create a server.
        let server = CommandQueueServerHandle::new(NAME);
        assert!(server.is_ok());

        // Create a client.
        let client = CommandQueueClientHandle::new(NAME);
        assert!(client.is_ok());
        let client = client.unwrap();

        // Send commands except for `Connect` command from client to server.
        assert!(matches!(
            client.paths_attach(PATHS),
            Err(Error::InvalidClient { pid: 0 })
        ));

        assert!(matches!(
            client.synchronize(
                PATH_0,
                &CacheTableKeyRef {
                    guest_addr: 0x400000,
                    guest_size: GUEST_CODE_0.len(),
                    guest_sign: &GUEST_CODE_0.into(),
                },
                &CacheTableVal {
                    host_meta: &HOST_META_0,
                    host_code: &HOST_CODE_0,
                },
            ),
            Err(Error::InvalidClient { pid: 0 })
        ));
    }

    #[test]
    fn unregistered_client_error() {
        const NAME: &str = "test-unregistered-client";

        // Create a server.
        let server = CommandQueueServerHandle::new(NAME);
        assert!(server.is_ok());
        let mut server = server.unwrap();

        // Create a client.
        let client = CommandQueueClientHandle::new(NAME);
        assert!(client.is_ok());
        let client = client.unwrap();

        let client_handle = thread::spawn(move || {
            // Send a command from client to server.
            assert!(client
                .queue
                .send(
                    Command::new(CommandKind::Synchronize {
                        path_len: PATH_0.len(),
                        cache_key: CacheTableKey::from_ref(&CacheTableKeyRef {
                            guest_addr: 0x400000,
                            guest_size: GUEST_CODE_0.len(),
                            guest_sign: &GUEST_CODE_0.into(),
                        }),
                        host_meta_size: HOST_META_0.len(),
                        host_code_size: HOST_CODE_0.len(),
                    })
                    .with_client_id(0)
                    .with_payload(
                        client.queue.mq_msgsize,
                        [PATH_0.as_bytes(), &HOST_META_0, &HOST_CODE_0]
                    )
                    .unwrap(),
                )
                .is_ok());
        });

        let server_handle = thread::spawn(move || {
            // Listen to the commands from client.
            let result = server
                .listen(Ok(()))
                .map_err(|error| error.downcast_ref::<Error>().cloned().unwrap());
            assert!(matches!(result, Err(Error::InvalidClient { pid: 0 })));
        });

        // Wait for the threads to finish.
        assert!(client_handle.join().is_ok());
        assert!(server_handle.join().is_ok());
    }

    #[test]
    fn server_handler_error() {
        const NAME: &str = "test-server-handler-error";

        // Create a server.
        let server = CommandQueueServerHandle::new(NAME);
        assert!(server.is_ok());
        let mut server = server.unwrap();

        // Create a client.
        let client = CommandQueueClientHandle::new(NAME);
        assert!(client.is_ok());
        let mut client = client.unwrap();

        let client_handle = thread::spawn(move || {
            // Send commands from client to server.
            assert!(client.connect().is_ok());
        });

        let server_handle = thread::spawn(move || {
            // Listen to the commands from client.
            let result = server
                .listen(Err(Error::InvalidCommand {
                    msg: "handle error".to_string(),
                }))
                .map_err(|error| error.downcast_ref::<Error>().cloned().unwrap());
            assert!(matches!(result, Err(Error::InvalidCommand { msg }) if msg == "handle error"));
        });

        // Wait for the threads to finish.
        assert!(client_handle.join().is_ok());
        assert!(server_handle.join().is_ok());
    }

    #[test]
    fn display() {
        let command = Command::new(CommandKind::Connect).with_client_id(0);
        assert_eq!(format!("{}", command), "Command from #0: Connect");

        let command = Command::new(CommandKind::Disconnect).with_client_id(0);
        assert_eq!(format!("{}", command), "Command from #0: Disconnect");

        let command = Command::new(CommandKind::PathsAttach {
            paths_len: PATH_0.len(),
        })
        .with_client_id(0);
        assert_eq!(
            format!("{}", command),
            format!(
                "Command from #0: PathsAttach ( paths-len: {} )",
                PATH_0.len()
            )
        );

        let command = Command::new(CommandKind::Synchronize {
            path_len: PATH_0.len(),
            cache_key: CacheTableKey::from_ref(&CacheTableKeyRef {
                guest_addr: 0x400000,
                guest_size: GUEST_CODE_0.len(),
                guest_sign: &GUEST_CODE_0.into(),
            }),
            host_meta_size: HOST_META_0.len(),
            host_code_size: HOST_CODE_0.len(),
        })
        .with_client_id(0);
        assert_eq!(
            format!("{}", command),
            format!(
                "Command from #0: Synchronize ( path: {} B; guest code / sign: @{:#X}-{:#X} / {}; host meta / code: {} B / {} B)",
                PATH_0.len(),
                0x400000,
                0x400000 + GUEST_CODE_0.len(),
                CodeSign::from(GUEST_CODE_0),
                HOST_META_0.len(),
                HOST_CODE_0.len(),
            )
        );

        let command = Command::new(CommandKind::Terminate {
            status: TerminateStatus::Success,
        })
        .with_client_id(0);

        assert_eq!(
            format!("{}", command),
            "Command from #0: Terminate ( Success )"
        );

        let command = Command::new(CommandKind::Terminate {
            status: TerminateStatus::Failure,
        })
        .with_client_id(0);

        assert_eq!(
            format!("{}", command),
            "Command from #0: Terminate ( Failure )"
        );
    }
}
